#include "EutranTcpReceiver.h"
#include "eutranManager.h"
#include "BbuManager.h"

#include <arpa/inet.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>

EutranTcpReceiver::EutranTcpReceiver():
maxFd_(-1),
serverFd_(-1)
{
}

EutranTcpReceiver* EutranTcpReceiver::instance()
{
    static EutranTcpReceiver instance_;
    return &instance_;
}

bool EutranTcpReceiver::initialize()
{
    if( !initServerFd() )
    {
        return false;
    }

    FD_ZERO(&readFdSet_);
    FD_ZERO(&writeFdSet_);
    FD_ZERO(&exceptionFdSet_);
    return true;
}

bool EutranTcpReceiver::initServerFd()
{
    struct sockaddr_in localAddr;
    localAddr.sin_family = AF_INET;
    char* envString = getenv("x86Ipaddr");
    if( envString!=NULL )
    {
        localAddr.sin_addr.s_addr = inet_addr(envString);
    }
    else
    {
        localAddr.sin_addr.s_addr = inet_addr("127.0.0.1");
    }
    envString = getenv("x86port");
    if( envString!=NULL )
    {
        localAddr.sin_port = htons(atoi(envString));
    }
    else
    {
        localAddr.sin_port = htons(32790);
    }

    dbgprintf(DEBUG_NOTICE, stdout, "ip is 0x%x; port is %d\n", localAddr.sin_addr.s_addr, ntohs(localAddr.sin_port));
    
    serverFd_= socket(AF_INET, SOCK_STREAM, 0);
    if( serverFd_==-1 )
    {
        dbgprintf(DEBUG_ERROR, stderr, "Fail to create socket, errno[%d] \n", errno);
        return false;
    }

    if( bind(serverFd_, (struct sockaddr *)&localAddr, sizeof(localAddr))==-1 )
    {
        dbgprintf(DEBUG_ERROR, stderr, "Fail to bind socket, errno[%d: %s] \n", errno, strerror(errno));
        return false;
    }

    int32_t listenRC = ::listen(serverFd_, 10000);
    if (listenRC == -1)
    {
        dbgprintf(DEBUG_ERROR, stderr, "listen() failed on fd=%d with error[%d]='%s' \n", serverFd_, errno, strerror(errno));
        return false;
    }

    struct linger lingerbuf;  
    lingerbuf.l_onoff = 1; 
    lingerbuf.l_linger = 0; 

    int ret = -1;
    do
    {
        ret = setsockopt(serverFd_, SOL_SOCKET, SO_LINGER, (char*)&lingerbuf, sizeof(lingerbuf));
    } while ((ret==-1) && (errno==EINTR));
        
    if (ret == -1) {
        EMLogError("setsockopt() for SO_LINGER failed on fd=%d with error[%d]='%s'",
        serverFd_,
        errno,
        strerror(errno));
    }

    maxFd_ = serverFd_;
    FD_SET(serverFd_, &readFdSet_);
    FD_SET(serverFd_, &writeFdSet_);
    dbgprintf(DEBUG_NOTICE, stdout, "serverFd=%d \n", serverFd_);
    return true;
}

int32_t EutranTcpReceiver::getServerFd()
{
    return serverFd_;
}

int32_t EutranTcpReceiver::getMaxFd()
{
    return maxFd_;
}

int32_t EutranTcpReceiver::getFd(in_addr_t addr)
{
    std::map<in_addr_t, int32_t>::iterator it = clientFds_.find(addr);
    if( it==clientFds_.end() )
    {
        return -1;
    }

    return it->second;
}

fd_set EutranTcpReceiver::getReadFdSet()
{
    FD_ZERO(&readFdSet_);
    FD_SET(serverFd_, &readFdSet_);

    for(std::map<in_addr_t, int32_t>::iterator it=clientFds_.begin(); it!=clientFds_.end(); it++)
    {
        FD_SET(it->second, &readFdSet_);
    }
    return readFdSet_;
}

fd_set EutranTcpReceiver::getWriteFdSet()
{
    FD_ZERO(&writeFdSet_);

    for(std::set<int32_t>::iterator it=writableFds_.begin(); it!=writableFds_.end(); it++)
    {
        FD_SET(*it, &writeFdSet_);
    }
    return writeFdSet_;
}

fd_set EutranTcpReceiver::getExceptionFdSet()
{
    FD_ZERO(&exceptionFdSet_);
    FD_SET(serverFd_, &exceptionFdSet_);

    for(std::map<in_addr_t, int32_t>::iterator it=clientFds_.begin(); it!=clientFds_.end(); it++)
    {
        FD_SET(it->second, &exceptionFdSet_);
    }
    return exceptionFdSet_;
}

void EutranTcpReceiver::handleReadEvent(int32_t fd)
{
    if( fd==serverFd_ )
    {
        int32_t clientFd = -1;
        struct sockaddr_in peerAddr;
        socklen_t len = sizeof(peerAddr);
        memset(&peerAddr, 0, len);
        
        do
        {
            clientFd = ::accept(serverFd_, (struct sockaddr *)&peerAddr, &len);
        } while ((clientFd==-1) && (errno==EINTR));

        std::map<in_addr_t, int32_t>::iterator oldIt = clientFds_.find(peerAddr.sin_addr.s_addr);
        if( oldIt!=clientFds_.end() )
        {
            EMLogWarning("Weird, BBU[addr:0x%x] socket not clean in time", peerAddr.sin_addr.s_addr);
            shutdown(oldIt->second);
            clientFds_.erase(oldIt);
        }
        clientFds_[peerAddr.sin_addr.s_addr] = clientFd;

        EMLogNotice("New connection with fd[%d] addr[0x%x]", clientFd, peerAddr.sin_addr.s_addr);

        struct linger lingerbuf;  
        lingerbuf.l_onoff = 1; 
        lingerbuf.l_linger = 0; 

        int ret = -1;
        do
        {
            ret = setsockopt(clientFd, SOL_SOCKET, SO_LINGER, (char*)&lingerbuf, sizeof(lingerbuf));
        } while ((ret==-1) && (errno==EINTR));
        
        if (ret == -1) {
            EMLogError("setsockopt() for SO_LINGER failed on fd=%d with error[%d]='%s'",
                clientFd,
                errno,
                strerror(errno));
        }

        int flags = fcntl(clientFd, F_GETFL, 0);
        do
        {
            ret = fcntl(clientFd, F_SETFL, flags|O_NONBLOCK);
        } while ((ret==-1) && (errno==EINTR));

        if (ret == -1)
        {
            EMLogError("ioctl() failed on fd=%d with errno[%d]='%s'",
                clientFd,
                errno,
                strerror(errno));
        }

        std::string recvBuffer;
        recvBuffer_[clientFd] = recvBuffer;
        clientAddr_[clientFd] = peerAddr.sin_addr.s_addr;

        std::string sendBuffer;
        sendBuffer_[clientFd] = sendBuffer;
        
        if( clientFd>maxFd_ )
        {
            maxFd_ = clientFd;
        }

        FD_SET(clientFd, &readFdSet_);
        FD_SET(clientFd, &writeFdSet_);
        return;
    }

    uint8_t recvMsgBuf[65535] = {0};
    uint8_t internalMsgBuffer[65535] = {0};
    int recv_len = -1;
    do
    {
        recv_len = ::recv(fd, recvMsgBuf, sizeof(recvMsgBuf), 0);
    }
    while ( (recv_len == -1) && (errno == EINTR) );

    if( recv_len<=0 )
    {
        EMLogInfo("recv len is 0 for FD[%d] with errno[%d]='%s'", fd,  errno,  strerror(errno));
        shutdown(fd);
        return;
    }
    std::map<int32_t, std::string>::iterator it = recvBuffer_.find(fd);
    if( it==recvBuffer_.end() )
    {
        EMLogError("Invalid fd[%d]", fd);
        return;
    }

    std::map<int32_t, in_addr_t>::iterator addrIt = clientAddr_.find(fd);
    if( addrIt==clientAddr_.end() )
    {
        EMLogError("Invalid fd[%d]", fd);
        return;
    }

    std::string& recvBuf = it->second;
    recvBuf.append(reinterpret_cast<char*>(&recvMsgBuf[0]), recv_len);

    if( recvBuf.length()<4 )
    {
        return;
    }

    uint16_t msgLength = 0;
    memcpy(&msgLength, &recvBuf[2], 2);
    msgLength = ntohs(msgLength);
    uint32_t totalMsgLength = 2 + sizeof(uint16_t) + msgLength;

    while( totalMsgLength<=recvBuf.length() )
    {
        std::string tmpBuf;
        tmpBuf.assign(recvBuf.c_str(), totalMsgLength);
        recvBuf.erase(0, totalMsgLength);
        
        uint8_t* data = reinterpret_cast<uint8_t*>(const_cast<char*>(tmpBuf.c_str()));
        EutranMgrPtr->processBbuMessage(data, totalMsgLength, addrIt->second);    

        if( clientAddr_.find(fd)==clientAddr_.end() )
        {
            break;
        }

        if( recvBuf.empty() )
        {
            break;
        }

        if( recvBuf.length()<4 )
        {
            break;
        }

        memcpy(&msgLength, &recvBuf[2], 2);
        msgLength = ntohs(msgLength);
        totalMsgLength = 2 + sizeof(uint16_t) + msgLength;
    }
}

void EutranTcpReceiver::handleReadEvents(fd_set* readFds)
{
    if( readFds==NULL )
    {
        EMLogError("NULL pointer");
        return;
    }
    if( FD_ISSET(serverFd_, readFds) )
    {
        handleReadEvent(serverFd_);
    }

    std::map<in_addr_t, int32_t> tmpFds = clientFds_;
    for(std::map<in_addr_t, int32_t>::iterator it=tmpFds.begin(); it!=tmpFds.end(); it++)
    {
        if( FD_ISSET(it->second, readFds) )
        {
            handleReadEvent(it->second);
        }
    }
}

void EutranTcpReceiver::handleWriteEvents(fd_set* writeFds)
{
    if( writeFds==NULL )
    {
        EMLogError("NULL pointer");
        return;
    }

    for(std::map<in_addr_t, int32_t>::iterator it=clientFds_.begin(); it!=clientFds_.end(); it++)
    {
        if( FD_ISSET(it->second, writeFds) )
        {
            if( !sendBuffer(it->second) )
            {
                EMLogError("size of clientFds_ is %d", clientFds_.size());
            }
        }
    }
}

bool EutranTcpReceiver::sendBuffer(int32_t fd)
{
    std::map<int32_t, std::string>::iterator sndBufIt = sendBuffer_.find(fd);
    if( sndBufIt==sendBuffer_.end() )
    {
        EMLogError("Fail to found send buffer from FD[%d]", fd);
        return false;
    }
        
    std::string& buf = sndBufIt->second;
    if( buf.empty() )
    {
        writableFds_.erase(fd);
        return true;
    }

    std::map<int32_t, in_addr_t>::iterator it=clientAddr_.find(fd);
    if( it==clientAddr_.end() )
    {
        EMLogError("Fail to found addr from FD[%d]", fd);
        return false;
    }

    EMLogDebug("BBU[0x%x]: send buffer is %d", it->second, buf.size());
    
    uint8_t* msg = reinterpret_cast<uint8_t*>(const_cast<char*>(buf.c_str()));
    uint32_t msgSize = buf.length();
    
    int send_rc = -1;
    do
    {
        send_rc = ::send(fd, msg, msgSize, MSG_DONTWAIT);
    }while ( (send_rc == -1) && (errno == EINTR) );
    
    if( send_rc==0 )
    {
        std::map<int32_t, in_addr_t>::iterator addrIt = clientAddr_.find(fd);
        if( addrIt==clientAddr_.end() )
        {
            EMLogError("Fail to found BBU addr from FD[%d]", fd);
            return false;
        }
        EMLogError("BBU[0x%x]: Fail to send message with error[%d]='%s''", addrIt->second, errno, strerror(errno));
        return false;
    }

    if( send_rc > msgSize )
    {
        EMLogError("BBU[0x%x]: sizeOfSentBuf[%d] is greater than sizeOfSendBuf[%d]", send_rc, msgSize);
        return false;
    }

    buf.erase(0, send_rc);
    if( buf.empty() )
    {
        writableFds_.erase(fd);
    }
    return true;
}

void EutranTcpReceiver::handleExcepEvents(fd_set* excepFds)
{
    if( excepFds==NULL )
    {
        EMLogError("NULL pointer");
        return;
    }

    if( FD_ISSET(serverFd_, excepFds) )
    {
        EMLogError("server fd exception");
        int32_t closeRC = 0;
        do
        {
            closeRC = ::close(serverFd_);
        } while ((closeRC==-1) && (errno==EINTR));

        initServerFd();
    }

    std::map<in_addr_t, int32_t> fds = clientFds_;
    std::map<in_addr_t, int32_t>::iterator it=fds.begin();
    while( it!=fds.end() )
    {
        if( FD_ISSET(it->second, excepFds) )
        {
            shutdown(it->second);
            clientFds_.erase(it++);
        }
        else
        {
            it++;
        }
    }
}

void EutranTcpReceiver::shutdown(in_addr_t bbuAddr)
{
    std::map<in_addr_t, int32_t>::iterator it = clientFds_.find(bbuAddr);

    if( it!=clientFds_.end() )
    {
        shutdown(it->second);
    }
}

void EutranTcpReceiver::shutdown(int32_t fd)
{
    EMLogInfo("shutdown conn[fd=%d]", fd);

    writableFds_.erase(fd);
    
    std::map<int32_t, std::string>::iterator sndBufIt = sendBuffer_.find(fd);
    if( sndBufIt!=sendBuffer_.end() )
    {
        sendBuffer_.erase(sndBufIt);
    }
    
    std::map<int32_t, std::string>::iterator fit = recvBuffer_.find(fd);
    if( fit!=recvBuffer_.end() )
    {
        recvBuffer_.erase(fit);
    }
    in_addr_t addrKey = 0;
    std::map<int32_t, in_addr_t>::iterator fit2 = clientAddr_.find(fd);
    if( fit2!=clientAddr_.end() )
    {
        addrKey = fit2->second;
        clientAddr_.erase(fit2);
    }

    std::map<in_addr_t, int32_t>::iterator fit3 = clientFds_.find(addrKey);
    if( fit3!=clientFds_.end() )
    {
        BbuCtrl* bbu = BbuManagerPtr->getBbu(fit3->first);
        if( bbu==NULL )
        {
            EMLogError("Fail to found BBU from addr[0x%x]", fit3->first);
        }
        else
        {
            EMLogWarning("BBU[%d] is down, remove fd[%d]", bbu->getInstance().getSlotId(), fit3->second);
        }
        bbu->indicateDeactive(false);
        clientFds_.erase(fit3);
    }

    int32_t closeRC = 0;
    do
    {
        closeRC = ::close(fd);
    } while ((closeRC==-1) && (errno==EINTR));

    if( closeRC!=0 )
    {
        EMLogError("close fd[%d] error[%d]='%s'", fd, errno, strerror(errno));
    }
}

bool EutranTcpReceiver::sendMessage(in_addr_t bbuAddr, uint8_t* msgBuf, uint32_t msgSize)
{
    int fd = EutranTcpReceiverPtr->getFd(bbuAddr);
    if( fd==-1 )
    {
        EMLogError("Invalid fd");
        return false;
    }

    std::map<int32_t, std::string>::iterator sndBufIt = sendBuffer_.find(fd);
    if( sndBufIt==sendBuffer_.end() )
    {
        EMLogError("Fail to found send buffer from FD[%d]", fd);
        return false;
    }
        
    std::string& buf = sndBufIt->second;
    if( !buf.empty() )
    {
        buf.append(reinterpret_cast<char*>(msgBuf), 0, msgSize);
        writableFds_.insert(fd);
        return true;
    }
    
    int send_rc = -1;
    do
    {
        send_rc = ::send(fd, msgBuf, msgSize, MSG_DONTWAIT);
    }while ( (send_rc == -1) && (errno == EINTR) );

    if( send_rc<0 )
    {
        EMLogError("BBU[0x%x]: Fail to send message with error[%d]='%s''", bbuAddr, errno, strerror(errno));
        return false;
    }

    if( send_rc==0 )
    {
        EMLogError("BBU[0x%x]: Fail to send message with error[%d]='%s''", bbuAddr, errno, strerror(errno));
        return false;
    }

    if( send_rc>msgSize )
    {
        EMLogError("BBU[0x%x]: sizeOfSentBuf[%d] is greater than msgSize[%d]", send_rc, msgSize);
        return false;
    }

    if( send_rc < msgSize )
    {
        char* bufPtr = reinterpret_cast<char*>(msgBuf);
        buf.assign(&bufPtr[send_rc], msgSize-send_rc);
        writableFds_.insert(fd);
    }
    return true;
}